
package com.leezai.mqtt.subsrict;

import com.leezai.mqtt.config.Properties;
import com.leezai.mqtt.dto.MessageWrapper;
import com.leezai.mqtt.dto.TopicConfig;
import com.leezai.mqtt.handler.MessageHandler;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;


/**
 * 订阅topic
 */
public class MqttSubscriberClient implements MqttCallbackExtended {

    private static final Logger log = LoggerFactory.getLogger(MqttSubscriberClient.class);

    private IMqttAsyncClient mqttClient;

    private final Properties properties;

    private final MessageHandler messageHandler;

    // 构造函数注入依赖
    public MqttSubscriberClient(MessageHandler messageHandler) {
        this.properties = Properties.getInstance();
        ;
        this.messageHandler = messageHandler;
    }


    public void connect() {
        try {
            MemoryPersistence persistence = new MemoryPersistence();
            UUID uuid = UUID.randomUUID();
            String clientId = properties.getClientIdPrefix() + uuid.toString();
            mqttClient = new MqttAsyncClient(properties.getUrl(), clientId, persistence);

            MqttConnectOptions options = new MqttConnectOptions();
            options.setUserName(properties.getUsername());
            options.setPassword(properties.getPassword().toCharArray());
            options.setAutomaticReconnect(true); // 自动重连
            options.setCleanSession(false);  // 保持持久会话（关键！）
            options.setKeepAliveInterval(60); // 心跳间隔（秒）
            options.setConnectionTimeout(30); // 连接超时（秒）
            options.setMaxReconnectDelay(5000); // 最大重连间隔（毫秒）

            mqttClient.setCallback(this);

            mqttClient.connect(options, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    List<TopicConfig> topics = properties.getTopics();
                    List<String> topicList = topics.stream().map(TopicConfig::getTopic).collect(Collectors.toList());
                    log.info("MQTT 客户端连接成功，开始订阅主题: " + topicList);

                    subscribeToTopics();
                    // 启动队列处理线程
                    messageHandler.startProcessing();
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    log.error("MQTT 连接失败: " + exception.getMessage());
                }
            });
        } catch (MqttException e) {
            log.error("leezai.mqtt MQTT连接异常 " + e.getMessage());
        }
    }

    @Override
    public void connectionLost(Throwable cause) {
        // 记录断开原因（重要调试信息）
        log.error("连接断开，原因: " + cause.getMessage() +
                "\n错误类型: " + (cause instanceof MqttException ?
                ((MqttException) cause).getReasonCode() : "非MQTT异常"));
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        messageHandler.addMessage(new MessageWrapper(topic, message.getPayload()));
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // 订阅客户端一般不关心这个回调
    }


    @Override
    public void connectComplete(boolean reconnect, String s) {
        if (reconnect) {
//            System.out.println("[自动重连成功] 开始恢复订阅...");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            subscribeToTopics();
        }
    }

    private void subscribeToTopics() {
        try {
            List<TopicConfig> topics = properties.getTopics();
            String[] topicArray = topics.stream().map(TopicConfig::getTopic).toArray(String[]::new);
            int[] qosArray = topics.stream().mapToInt(TopicConfig::getQos).toArray();
            mqttClient.subscribe(topicArray, qosArray, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    log.info("成功订阅主题: " + Arrays.toString(topicArray));
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    log.error("订阅失败: " + exception.getMessage());
                }
            });
        } catch (MqttException e) {
            log.error("订阅异常: " + e.getMessage());
        }
    }

    // 关闭方法
    public void disconnect() {
        try {
            if (mqttClient != null) {
                mqttClient.disconnect();
                mqttClient.close();
            }
        } catch (MqttException e) {
            log.error("关闭MQTT客户端异常: " + e.getMessage());
        } finally {
            // 确保消息处理器资源被释放
            try {
                messageHandler.shutdown();
            } catch (Exception e) {
                log.error("关闭消息队列处理器异常: " + e.getMessage(), e);
            }
        }
    }
}